package com.aaa.consumer;

import com.aaa.entity.Info;
import com.aaa.service.InfoService;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @author ：Teacher陈（86521760@qq.com）
 * @date ：Created in 2022/5/20 10:33
 * @description：消息的消费者
 * @modified By：
 * @version: 1.0
 */
@Component
@Slf4j
public class MyConsumer {
    /**
     * 服务对象
     */
    @Resource
    private InfoService infoService;
    /**
     * 去消息队列中取数据
     * 什时候取？依靠监听器
     */
    @KafkaListener(topics = "mytopic")
   public void  getInfo(ConsumerRecord record){
        //获取消息队列中指定主题的消息
        Object value = record.value();
       /* log.info("====="+value.toString());*/
        //将消费的消息转换成java实体类对象
        //jackson的工具类，springboot自带
        ObjectMapper objectMapper = new ObjectMapper();
        Info info = null;
        try {
            info = objectMapper.readValue(value.toString(), Info.class);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        log.info("消费者打印消息："+info.toString());
        //将java实体对象保存到数据库中(下次课)
        infoService.insert(info);
    }
}
